package utils.db;

import java.io.IOException;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Date;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

public class dbToHbase {
	
	private static Configuration config = HBaseConfiguration.create();
	/**
	 * Store opinions in hbase
	 * @throws IOException
	 * @throws ClassNotFoundException
	 * @throws SQLException
	 */
	public static void storeOpinions() throws IOException, ClassNotFoundException, SQLException{
		int start = 436999;
		Date current = null;
		byte[] currentDate = null;
		byte[] topicFamily = Bytes.toBytes("topicId");
		
		HTable tableOpinions = new HTable(config, "opinions");
		HTable tableDate = new HTable(config, "date");

		while(true){
			ResultSet res = utils.db.Tools.executeSelect("trends","select OPINION_ID,DATE_ENTERED,TOPIC_ID,BODY"+
					" from opinions where LANGUAGE_ID = 1 and OPINION_ID >="+start
			+ " order by OPINION_ID asc limit 100000;");
			int opinionId = 0;
			while(res.next()){
				current = res.getDate("DATE_ENTERED");
				currentDate = Bytes.toBytes(current.toString());
				Put docPut = new Put(Bytes.toBytes(current.toString()));
				docPut.add(topicFamily, Bytes.toBytes(res.getString("TOPIC_ID")), Bytes.toBytes(res.getString("BODY")));
				tableOpinions.put(docPut);
				tableDate.incrementColumnValue(currentDate, Bytes.toBytes("date"), currentDate, 1);
				
				opinionId = Integer.parseInt(res.getString("OPINION_ID"));
				start = opinionId + 1;
				System.out.println("Get and Store From MySql in storeOpinions ID: = " + opinionId);
			}
			if(opinionId >= 18281465)
				break;
		}
		tableDate.close();
		tableOpinions.close();
	}
	
	/**
	 * store topic in hbase
	 * @throws IOException
	 * @throws ClassNotFoundException
	 * @throws SQLException
	 */
	public static void storeTopic() throws IOException, ClassNotFoundException, SQLException{
		int id = 0;
		Date current = null;
		byte[] currentDate = null;
		byte[] dateFamily = Bytes.toBytes("date");
		byte[] topicFamily = Bytes.toBytes("topicId");
		
		HTable tableTopic = new HTable(config, "topic");
		HTable tableDateToTopic = new HTable(config, "dateToTopic");
		HTable tableDate = new HTable(config, "date");
		ResultSet res = utils.db.Tools.executeSelect("trends","select TOPIC_ID,TITLE,DATE_ENTERED"+
				" from topics where LANGUAGE_ID = 1;");

		
		while(res.next()){
			current = res.getDate("DATE_ENTERED");
			currentDate = Bytes.toBytes(current.toString());
			
			Put docPut = new Put(Bytes.toBytes(res.getString("TOPIC_ID")));
			Put docPutDate = new Put(currentDate);
			
			docPut.add(dateFamily, currentDate, Bytes.toBytes(res.getString("TITLE")));
			docPutDate.add(topicFamily,Bytes.toBytes(res.getString("TOPIC_ID")), Bytes.toBytes(res.getString("TITLE")));
			
			tableTopic.put(docPut);
			tableDateToTopic.put(docPutDate);
			
			tableDate.incrementColumnValue(currentDate, Bytes.toBytes("date"), currentDate, 1);
			
			id++;
			System.out.println("Get and Store From MySql element: " + id + " in storeTopic");
		}
		
		tableTopic.close();
		tableDate.close();
		tableDateToTopic.close();
	}
	
	
	/**
	 * store polls in hbase
	 * @throws IOException
	 * @throws ClassNotFoundException
	 * @throws SQLException
	 */
	public static void storePolls() throws IOException, ClassNotFoundException, SQLException{
		int id = 0;
		Date current = null;
		byte[] currentDate = null;
		byte[] pollIdFamily = Bytes.toBytes("pollId");
		
		HTable tablePolls = new HTable(config, "polls");

		ResultSet res = utils.db.Tools.executeSelect("trends","select PollID, DateOpen, Question from trends.polls "+
				" where PollLanguage = 1");

		
		while(res.next()){
			current = res.getDate("DateOpen");
			currentDate = Bytes.toBytes(current.toString());
			
			Put docPut = new Put(currentDate);
			docPut.add(pollIdFamily, Bytes.toBytes(res.getString("PollID")), Bytes.toBytes(res.getString("Question")));
			tablePolls.put(docPut);
			id++;
			System.out.println("Get and Store From MySql element: " + id + " in storePolls");
		}
		
		tablePolls.close();
	}
}
